package com.asap.demo.count;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.util.Collector;

/**
 * @author wangbh
 * @Description: 1
 * @date 2021/6/9 17:13
 */
public class WordWithCount {
//java -Djava.ext.dirs=/home/asap/flink-1.13.1/lib -cp flink-test-1.0-SNAPSHOT.jar com.WordWithCount --ip 192.168.1.25 --port 8081 --input /home/asap/wbh/json/nohup.out
    public static void main(String[] args) throws Exception {


        final ParameterTool params = ParameterTool.fromArgs(args);
        final int port;
        final String ip;
        DataSet<String> text;
        try {
            ip = params.get("ip");
            port = params.getInt("port");
            final ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment(ip, port, 2);
            text = env.readTextFile(params.get("input"));
            DataSet<Tuple2<String, Integer>> counts =
                    // split up the lines in pairs (2-tuples) containing: (word,1)
                    text.flatMap(new Tokenizer())
                            // group by the tuple field "0" and sum up tuple field "1"
                            .groupBy(0)
                            .sum(1);

            System.out.println("Printing result to stdout. Use --output to specify output path.");
            counts.print();
            env.execute();
            System.out.println("Printing result to stdout. Use --output to specify output path.");

        } catch (Exception e) {
            System.err.println("No port or input or ip specified. Please run 'SocketWindowWordCount --ip <ip> --port <port>'" +
                    " --input <input>");
            return;
        }


    }
// *************************************************************************
//     USER FUNCTIONS
// *************************************************************************

    /**
     * Implements the string tokenizer that splits sentences into words as a user-defined
     * FlatMapFunction. The function takes a line (String) and splits it into
     * multiple pairs in the form of "(word,1)" ({@code Tuple2<String, Integer>}).
     */
    public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {

        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            // normalize and split the line
            String[] tokens = value.toLowerCase().split("\\W+");

            // emit the pairs
            for (String token : tokens) {
                if (token.length() > 0) {
                    out.collect(new Tuple2<String, Integer>(token, 1));
                }
            }
        }
    }
}
